package com.atguigu.gmall.realtime.app.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.constant.GmallConfig;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    MapStateDescriptor<String, TableProcess> tableProcessStateDesc = null;

    // 构造器
    public TableProcessFunction(MapStateDescriptor<String, TableProcess> tableProcessMapStateDesc) {
        this.tableProcessStateDesc = tableProcessMapStateDesc;
    }

    // 处理主流数据如何使用广播状态
    //1  根据状态筛选数据，只保留状态中标识的维度表数据
    //2  根据配置表，对原数据的字段进行剪裁，不在配置表中定义的字段一律去除
    //3  标识改维度数据要进入哪个phoenix的维度表，打个目标表标记，未来在sink要根据此标识插入对应表
    //4  维度数据纳入收集器
//{
//  "database": "gmall",
//  "xid": 598500,
//  "data": {
//    "birthday": "1977-05-10",
//    "create_time": "2020-06-10 11:35:07",
//    "login_name": "aeg1ou4yaxz",
//    "nick_name": "义兴44",
//    "name": "秦义兴",
//    "user_level": "1",
//    "phone_num": "13736483945",
//    "id": 3,
//    "email": "aeg1ou4yaxz@yahoo.com",
//    "operate_time": "2020-06-14 11:36:59"
//  },
//  "old": {
//    "nick_name": "义兴"
//  },
//  "commit": true,
//  "type": "update",
//  "table": "user_info",
//  "ts": 1668649193
//}
    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        System.out.println("主流数据：" + jsonObject);
        String table = jsonObject.getString("table");
        if (jsonObject.getString("type").equals("bootstrap-start") || jsonObject.getString("type").equals("bootstrap-complete")) {
            return;
        }
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(this.tableProcessStateDesc);
        TableProcess tableProcess = broadcastState.get(table);
        System.out.println("状态：" + tableProcess);
        //1  根据状态筛选数据，只保留状态中标识的维度表数据
        if (tableProcess != null) {  // 不为空 ，就说明是维度数据
            //2  根据配置表，对原数据的字段进行剪裁，不在配置表中定义的字段一律去除
            String sinkColumns = tableProcess.getSinkColumns();
            String[] sinkColArr = sinkColumns.split(",");
            HashSet<String> sinkColSet = new HashSet();
            Collections.addAll(sinkColSet, sinkColArr);
            //不在配置表中定义的字段一律去除
            JSONObject dataJsonObj = jsonObject.getJSONObject("data");
            Set<Map.Entry<String, Object>> dataEntryValue = dataJsonObj.entrySet();
            dataEntryValue.removeIf(entry -> !sinkColSet.contains(entry.getKey()));

            //3  标识改维度数据要进入哪个phoenix的维度表，打个目标表标记，未来在sink要根据此标识插入对应表
            jsonObject.put("sink_table", tableProcess.getSinkTable());
            //4 纳入收集器
            out.collect(jsonObject);
        }

    }

    // 处理广播流如何更新广播状态
    // 1  主要目的： 把配置的信息更新到广播状态中
    // 2  次要目的： 感知到配置信息的变化 ，维护hbase中维度的表结构 ，自动创建表  # 去掉 在广播前已经完成
    // {
    //  "before": null,
    //  "after": {
    //    "source_table": "base_dic",
    //    "sink_table": "dim_base_dic",
    //    "sink_columns": "id,name",
    //    "sink_pk": "id",
    //    "sink_extend": null
    //  },
    //  "source": {
    //    "version": "1.5.4.Final",
    //    "connector": "mysql",
    //    "name": "mysql_binlog_source",
    //    "ts_ms": 1668588518000,
    //    "snapshot": "false",
    //    "db": "gmall_realtime_config",
    //    "sequence": null,
    //    "table": "table_process",
    //    "server_id": 1,
    //    "gtid": null,
    //    "file": "mysql-binlog.000026",
    //    "pos": 1178236,
    //    "row": 0,
    //    "thread": null,
    //    "query": null
    //  },
    //  "op": "c",
    //  "ts_ms": 1668743587601,
    //  "transaction": null
    //}
    @Override
    public void processBroadcastElement(String tableProcessCdcJsonstr, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // System.out.println(" 配置流数据：" + tableProcessCdcJsonstr);
        // 1 把json字符串结构化
        JSONObject tableProcessCdcJsonobj = JSON.parseObject(tableProcessCdcJsonstr);
        //2  感知到配置信息的变化 ，维护hbase中维度的表结构 ，自动创建表, 增加维度字段
        TableProcess tableProcess = tableProcessCdcJsonobj.getObject("after", TableProcess.class);
        //   checkTable(tableProcess);

        //3 主要目的： 把配置的信息更新到广播状态中
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(this.tableProcessStateDesc);

        // 4 如果是删除配置则 从从状态中移除，否则保留更新状态
        if (tableProcessCdcJsonobj.getString("op").equals("d")) {
            broadcastState.remove(tableProcess.getSourceTable());
        } else {
            broadcastState.put(tableProcess.getSourceTable(), tableProcess);
        }
    }
}
